package com.bruce.codeFactory.Thread.ThreadPool;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.IntStream;

/**
 * @author luotuan
 * @Description 一个线程池的例子
 * @create 2019-02-20 21:55
 **/
public class SimpleThreadPool extends Thread {
    //线程池大小
    private int size;
    //队列大小
    private final int queueSize;
    //默认线程池大小
    private final static int DEFAULT_SIZE = 10;
    //默认队列大小
    private final static int DEFAULT_TASK_QUEUE_SIZE = 2000;
    private static volatile int seq = 0;
    private static final String THREAD_PREFIX = "SIMPLE_THREAD_POOL-";
    //默认线程组
    private static final ThreadGroup GROUP = new ThreadGroup("Pool_Group");
    //任务队列
    private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();
    //线程集合
    private static final List<WorkerTask> THREAD_QUEUE = new ArrayList<>();
    //拒绝策略
    private final DiscardPolicy discardPolicy;
    private final static DiscardPolicy DEFAULT_DISCARD_POLICY = () -> {
        throw new DiscardException("Discard This Task");
    };
    //线程池是否关闭
    private volatile boolean destroy = false;

    private int min;
    private int active;
    private int max;

    public SimpleThreadPool() {
        this(4, 8, 12, DEFAULT_TASK_QUEUE_SIZE, DEFAULT_DISCARD_POLICY);
    }

    public SimpleThreadPool(int min, int active, int max, int queueSize, DiscardPolicy discardPolicy) {
        this.min = min;
        this.active = active;
        this.max = max;
        this.queueSize = queueSize;
        this.discardPolicy = discardPolicy;
        init();
    }

    /**
     * 提交线程任务
     *
     * @param runnable
     */
    public void submit(Runnable runnable) {
        //线程池关闭状态下允许提交任务
        if (destroy) {
            throw new IllegalStateException("The thread pool already destroy and not allow submit task");
        }
        synchronized (TASK_QUEUE) {
            //队列任务数量大于设置的队列数时，拒绝
            if (TASK_QUEUE.size() >= queueSize) {
                discardPolicy.discard();
            }
            TASK_QUEUE.addLast(runnable);
            TASK_QUEUE.notifyAll();
        }
    }

    /**
     * 初始化线程池
     */
    private void init() {
        for (int i = 0; i < this.min; i++) {
            createWorkerTask();
        }
        this.size = min;
        this.start();
    }

    /**
     * 开启一个线程
     */
    private void createWorkerTask() {
        WorkerTask workerTask = new WorkerTask(GROUP, THREAD_PREFIX + (seq++));
        workerTask.start();
        THREAD_QUEUE.add(workerTask);
    }

    /**
     * 关闭线程池
     *
     * @throws InterruptedException
     */
    public void shutdown() throws InterruptedException {
        while (!TASK_QUEUE.isEmpty()) {
            Thread.sleep(50);
        }
        int initVal = THREAD_QUEUE.size();
        while (initVal > 0) {
            for (WorkerTask task : THREAD_QUEUE) {
                if (task.getTaskState() == TaskState.BLOCKED) {
                    task.interrupt();
                    task.close();
                    initVal--;
                } else {
                    Thread.sleep(10);
                }
            }
        }
        this.destroy = true;
        System.out.println("The thread pool disposed.");
    }

    public int getSize() {
        return size;
    }

    public int getQueueSize() {
        return queueSize;
    }

    public boolean isDestroy() {
        return this.destroy;
    }

    /**
     * 线程状态
     */
    private enum TaskState {
        FREE, RUNNING, BLOCKED, DEAD
    }

    public static class DiscardException extends RuntimeException {
        public DiscardException(String message) {
            super(message);
        }
    }

    public interface DiscardPolicy {
        void discard() throws DiscardException;
    }

    @Override
    public void run() {
        while (!destroy) {
            System.out.printf("min:%d,active:%d,max:%d,current:%d,queueSize:%d\n",
                    this.min, this.active, this.max, this.size, TASK_QUEUE.size());
            try {
                Thread.sleep(5000);
                if (TASK_QUEUE.size() > active && size < active) {
                    for (int i = size; i < active; i++) {
                        createWorkerTask();
                    }
                    System.out.println("The pool incremented to active");
                    size = active;
                } else if (TASK_QUEUE.size() > max && size < max) {
                    for (int i = size; i < max; i++) {
                        createWorkerTask();
                    }
                    System.out.println("The pool incremented to max");
                    size = max;
                }
                synchronized (THREAD_QUEUE) {
                    if (TASK_QUEUE.isEmpty() && size > active) {
                        int releaseSize = size - active;
                        for (Iterator<WorkerTask> iterator = THREAD_QUEUE.iterator(); iterator.hasNext(); ) {
                            if (releaseSize <= 0) {
                                break;
                            }
                            System.out.println("===reduce=====");
                            WorkerTask task = iterator.next();
                            task.close();
                            task.interrupt();
                            iterator.remove();
                            releaseSize--;
                        }
                        size = active;
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 线程类
     */
    private static class WorkerTask extends Thread {
        //线程运行状态
        private volatile TaskState taskState = TaskState.FREE;

        public WorkerTask(ThreadGroup group, String name) {
            super(group, name);
        }

        public TaskState getTaskState() {
            return this.taskState;
        }

        @Override
        public void run() {
            OUTER:
            while (this.taskState != TaskState.DEAD) {
                Runnable runnable;
                synchronized (TASK_QUEUE) {
                    //队列为空，线程等待
                    while (TASK_QUEUE.isEmpty()) {
                        try {
                            taskState = TaskState.BLOCKED;
                            TASK_QUEUE.wait();
                        } catch (InterruptedException e) {
                            //中断线程后线程退出
                            System.out.println(Thread.currentThread().getName() + " interrupted...");
                            break OUTER;
                        }
                    }
                    runnable = TASK_QUEUE.removeFirst();
                }
                if (runnable != null) {
                    taskState = TaskState.RUNNING;
                    runnable.run();
                    taskState = TaskState.FREE;
                }
            }
        }

        public void close() {
            this.taskState = TaskState.DEAD;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        SimpleThreadPool pool = new SimpleThreadPool();
        IntStream.rangeClosed(0, 80)
                .forEach(i -> {
                    pool.submit(() -> {
                        System.out.println("The runable " + i + " be serviced by " + Thread.currentThread() + " start.");
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("The runable " + i + " be serviced by " + Thread.currentThread() + " finished.");
                    });
                });
        Thread.sleep(20000);
        pool.shutdown();
    }
}
